package felix.rule.service.impl;

import felix.actor.core.util.ThingsRouterThreadFactory;
import felix.rule.api.RuleChainTransactionService;
import felix.rule.api.TransactionTask;
import felix.rule.data.msg.EngineMsg;
import felix.rule.service.executor.DbCallbackExecutorService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.Queue;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;

@Service
@Slf4j
public class BaseRuleChainTransactionService implements RuleChainTransactionService {

    private final DbCallbackExecutorService callbackExecutor;

    @Value("${actors.rule.transaction.queue_size:10}")
    private int finalQueueSize;

    private final Lock transactionLock = new ReentrantLock();
    private final ConcurrentMap<String, BlockingQueue<TransactionTask>> transactionMap = new ConcurrentHashMap<>();
    private final Queue<TransactionTask> timeoutQueue = new ConcurrentLinkedQueue<>();

    private ExecutorService timeoutExecutor;

    public BaseRuleChainTransactionService(DbCallbackExecutorService callbackExecutor) {
        this.callbackExecutor = callbackExecutor;
    }

    @PostConstruct
    public void init() {
        timeoutExecutor = Executors.newSingleThreadExecutor(ThingsRouterThreadFactory.forName("rule-chain-transaction"));
        executeOnTimeout();
    }

    @PreDestroy
    public void destroy() {
        if (timeoutExecutor != null) {
            timeoutExecutor.shutdownNow();
        }
    }

    @Override
    public void beginTransaction(EngineMsg msg, Long duration, Consumer<EngineMsg> onStart, Consumer<EngineMsg> onEnd, Consumer<Throwable> onFailure) {
        transactionLock.lock();
        try {
            BlockingQueue<TransactionTask> queue = transactionMap.computeIfAbsent(msg.getOriginator(), id ->
                    new LinkedBlockingQueue<>(finalQueueSize));

            TransactionTask transactionTask = new TransactionTask(msg, onStart, onEnd, onFailure, duration);
            int queueSize = queue.size();
            if (queueSize >= finalQueueSize) {
                log.trace("Queue has no space: {}", transactionTask);
                executeOnFailure(transactionTask.getOnFailure(), "Queue has no space!");
            } else {
                addMsgToQueues(queue, transactionTask);
                if (queueSize == 0) {
                    executeOnSuccess(transactionTask.getOnStart(), transactionTask.getMsg());
                } else {
                    log.trace("Msg [{}][{}] is waiting to start transaction!", msg.getId(), msg.getType());
                }
            }
        } finally {
            transactionLock.unlock();
        }
    }

    @Override
    public void endTransaction(EngineMsg msg, Consumer<EngineMsg> onSuccess, Consumer<Throwable> onFailure) {
        // 集群节点
//        Optional<ServerAddress> address = routingService.resolveById(msg.getTransactionData().getOriginatorId());
//        if (address.isPresent()) {
//            sendTransactionEventToRemoteServer(msg, address.get());
//            executeOnSuccess(onSuccess, msg);
//        } else {
        endLocalTransaction(msg, onSuccess, onFailure);
//        }
    }

    private void addMsgToQueues(BlockingQueue<TransactionTask> queue, TransactionTask transactionTask) {
        queue.offer(transactionTask);
        timeoutQueue.offer(transactionTask);
        log.trace("Added msg to queue, size: [{}]", queue.size());
    }

    private void endLocalTransaction(EngineMsg msg, Consumer<EngineMsg> onSuccess, Consumer<Throwable> onFailure) {
        transactionLock.lock();
        try {
            BlockingQueue<TransactionTask> queue = transactionMap.computeIfAbsent(msg.getOriginator(), id ->
                    new LinkedBlockingQueue<>(finalQueueSize));

            TransactionTask currentTransactionTask = queue.peek();
            if (currentTransactionTask != null) {
                if (currentTransactionTask.getMsg().getId().equals(msg.getId())) {
                    currentTransactionTask.setCompleted(true);
                    queue.poll();
                    log.trace("Removed msg from queue, size [{}]", queue.size());
                    // 前置节点完成
                    executeOnSuccess(currentTransactionTask.getOnEnd(), currentTransactionTask.getMsg());
                    // end 节点完成
                    executeOnSuccess(onSuccess, msg);

                    TransactionTask nextTransactionTask = queue.peek();
                    if (nextTransactionTask != null) {
                        executeOnSuccess(nextTransactionTask.getOnStart(), nextTransactionTask.getMsg());
                    }
                } else {
                    // 重新消费
                    log.trace("Task has expired!");
                    executeOnFailure(onFailure, "Task has expired!");
                }
            } else {
                log.trace("Queue is empty, previous task has expired!");
                executeOnFailure(onFailure, "Queue is empty, previous task has expired!");
            }
        } finally {
            transactionLock.unlock();
        }
    }

    private void executeOnTimeout() {
        timeoutExecutor.submit(() -> {
            while (true) {
                TransactionTask transactionTask = timeoutQueue.peek();
                if (transactionTask != null) {
                    long sleepDuration = 0L;
                    transactionLock.lock();
                    try {
                        if (transactionTask.isCompleted()) {
                            timeoutQueue.poll();
                        } else {
                            long expIn = transactionTask.getExpirationTime() - System.currentTimeMillis();
                            if (expIn < 0) {
                                log.trace("Task has expired! Deleting it...[{}][{}]", transactionTask.getMsg().getId(), transactionTask.getMsg().getType());
                                timeoutQueue.poll();

                                BlockingQueue<TransactionTask> queue = transactionMap.get(transactionTask.getMsg().getOriginator());
                                if (queue != null) {
                                    queue.poll();
                                    TransactionTask nextTransactionTask = queue.peek();
                                    if (nextTransactionTask != null) {
                                        executeOnFailure(transactionTask.getOnFailure(), "Task has expired!");
                                    }
                                }
                            } else {
                                sleepDuration = Math.min(expIn, transactionTask.getDuration());
                            }
                        }
                    } finally {
                        transactionLock.unlock();
                    }
                    if (sleepDuration > 0L) {
                        try {
                            log.trace("Task has not expired! Continue executing...[{}][{}]", transactionTask.getMsg().getId(), transactionTask.getMsg().getType());
                            TimeUnit.MILLISECONDS.sleep(sleepDuration);
                        } catch (InterruptedException e) {
                            throw new IllegalStateException("Thread interrupted", e);
                        }
                    }
                } else {
                    try {
                        log.trace("Queue is empty, waiting for tasks!");
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        throw new IllegalStateException("Thread interrupted", e);
                    }
                }
            }
        });
    }

    private void executeOnFailure(Consumer<Throwable> onFailure, String exception) {
        executeCallback(() -> {
            onFailure.accept(new RuntimeException(exception));
            return null;
        });
    }

    private void executeOnSuccess(Consumer<EngineMsg> onSuccess, EngineMsg engineMsg) {
        executeCallback(() -> {
            onSuccess.accept(engineMsg);
            return null;
        });
    }

    private void executeCallback(Callable<Void> task) {
        callbackExecutor.executeAsync(task);
    }

}
